package com.flying.rsocket;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;

import java.nio.ByteBuffer;

import static com.flying.rsocket.Constants.DATA_STREAM_NAME;
import static com.flying.rsocket.Constants.TCP_PORT;

public class ReqStreamClient {
    private final RSocket socket;

    public ReqStreamClient() {
        this.socket = RSocketFactory.connect()
                .transport(TcpClientTransport.create("localhost", TCP_PORT))
                .start()
                .block();
    }

    public Flux<Float> getDataStream() {
        return socket.requestStream(DefaultPayload.create(DATA_STREAM_NAME))
                .map(Payload::getData)
                .map(ByteBuffer::getFloat)
                .onErrorReturn(null);
    }

    public void dispose() {
        socket.dispose();
    }
}
